namespace chatroom{
    // 1  brpc服务端
    class MessageServiceImp : public chatroom::MessageService{
    public:
        MessageServiceImp(chatroom::MessageStorgeTable::Ptr sql_msg,
        chatroom::ESMessage::Ptr es_message,
        std::string user_service_name,
        std::string file_service_name,
        ServiceManage::Sptr mm_channel)
        :_sql_msg(sql_msg),_es_message(es_message)
        ,_user_service_name(user_service_name),_file_service_name(file_service_name),_mm_channel(mm_channel){}
        ~MessageServiceImp(){}
            virtual void GetHistoryMsg(::google::protobuf::RpcController* controller,
                       const ::chatroom::GetHistoryMsgReq* request,
                       ::chatroom::GetHistoryMsgRsp* response,
                       ::google::protobuf::Closure* done)
                       {
                            brpc::ClosureGuard guard(done);
                            auto errmsg_response = [this,response](const std::string& rid,const std::string& errmsg) ->void{
                                            response->set_success(false);
                                            response->set_request_id(rid);
                                            request->set_errmsg(errmsg);
                                            return;
                            }
                           std::string rid = request->request_id();
                           std::string chat_session_id = request->chat_session_id();
                           int64_t start_time = boost::posix_time::ptime::from_time_t(request->start_time());
                           int64_t over_time  = boost::posix_time::ptime::request->over_time();
                           //获取到消息列表
                           auto message_lists = _sql_msg->get_range_message(chat_session_id,start_time,over_time);
                            //获取用户信息
                            unorder_set<std::string> set1;
                            unordered_map<std::string,UserInfo> Hashmap1;
                            for(auto& msg : message_lists)
                            {
                                set1.insert(msg.get_send_user_id());
                            }
                            bool test = _user_info(rid,set1,Hashmap1);
                            if(!test)
                            {
                                Logger_ERROR("{}获取用户信息失败",rid);
                                return errmsg_response(rid,"获取用户信息失败");
                            }
                            //获取文件信息
                            unorder_set<std::string> set2;
                            unorder_set<std::string,std::string> Hashmap2;
                            for(auto& msg : message_lists)
                            {
                                if(msg.get_file_id().empty()) continue;
                                set2.insert(msg.get_file_id());
                            }
                            test = _Getfile(rid,set2,Hashmap2);
                            if(!test)
                            {
                                Logger_ERROR("{}获取文件信息失败",rid);
                                return errmsg_response(rid,"获取文件信息失败");
                            }
                            //组织响应
                            response->set_success(true);
                            response->set_request_id(rid);
                            for(auto& msg : message_lists)
                            {
                                auto message_info = response->add_msg_list();
                                message_info->set_message_id(uuid());
                                message_info->set_chat_session_id(chat_session_id);
                                message_info->set_timestamp(msg.get_created_time());
                                message_info->mutable_sender()->CopyFrom(Hashmap1[msg.get_send_user_id()]);
                                switch(msg.get_message_type())
                                {
                                    case MessageContent::MessageType::STRING:
                                        message_info->mutable_message()->set_message_type(MessageType::STRING);
                                        message_info->mutable_message()->mutable_string_message()->set_content(Hashmap2[msg.get_file_id()]);
                                        break;
                                    case MessageContent::MessageType::IMAGE:
                                        message_info->mutable_message()->set_message_type(MessageType::IMAGE);
                                        message_info->mutable_message()->mutable_image_message()->set_file_id(msg.get_file_id());
                                        message_info->mutable_message()->mutable_image_message()->set_image_content(msg.content());
                                        break;
                                    case MessageContent::MessageType::FILE:
                                        message_info->mutable_message()->set_message_type(MessageType::FILE);
                                        message_info->mutable_message()->mutable_file_message()->set_file_id(msg.get_file_id());
                                        message_info->mutable_message()->mutable_file_message()->set_file_size(msg.get_file_size());
                                        message_info->mutable_message()->mutable_file_message()->set_file_name(msg.get_file_name());
                                        message_info->mutable_message()->mutable_file_message()->set_file_contents(msg.get_content());
                                        break;
                                    case MessageContent::MessageType::SPEECH:
                                        message_info->mutable_message()->set_message_type(MessageType::SPEECH);
                                        message_info->mutable_message()->mutable_speech_message()->set_file_id(msg.get_file_id());
                                        message_info->mutable_message()->mutable_speech_message()->set_file_contents(msg.get_content());
                                        break;
                                    default:
                                        Logger_ERROR("{}未能识别到的消息",rid);
                                        return errmsg_response(rid,"未能识别到的消息");
                                }
                            }
                       }
            virtual void GetRecentMsg(::google::protobuf::RpcController* controller,
                       const ::chatroom::GetRecentMsgReq* request,
                       ::chatroom::GetRecentMsgRsp* response,
                       ::google::protobuf::Closure* done)
                       {
                            // 1. 从请求中，获取会话 ID， 要获取的消息数量
                            int64_t count = request->msg_count();
                            std::string rid = request->request_id();
                            std::string chat_session_id = request->chat_session_id();
                            // 2. 访问数据库，从数据库中按时间排序，获取指定数量的消息简略信息（消息 ID，
                            // 会话 ID，消息类型，产生时间，发送者用户 ID，文本消息内容，文件消息元信息）
                            auto messsge_lists = _sql_msg->get_recent_message(chat_session_id,count);
                            // 3. 循环构造完整消息（从用户子服务获取消息的发送者用户信息，从文件子服务获
                            // 取文件/语音/图片数据）
                            //获取用户信息
                            unorder_set<std::string> set1;
                            unordered_map<std::string,UserInfo> Hashmap1;
                            for(auto& msg:message_lists)
                            {
                                set1.insert(msg.get_send_user_id());
                            }
                            bool ret = _user_info(rid,set1,Hashmap1);
                            if(!ret)
                            {
                                Logger_ERROR("{}获取用户信息失败");
                                return errmsg_response(rid,"获取用户信息失败");
                            }
                            //获取文件相关信息
                            unorder_set<std::string> set2;
                            unordered_map<std::string,std::string> Hashmap2;
                            for(auto& msg : messsge_lists)
                            {
                                set1.insert(msg.get_file_id());
                            }
                            ret = _Getfile(rid,set2,Hashmap2);
                            if(!ret)
                            {
                                Logger_ERROR("{}获取文件信息失败");
                                return errmsg_response(rid,"获取文件信息失败");
                            }
                            // 4. 组织响应返回给网关服务器。
                            response->set_success(true);
                            response->set_request_id(rid);
                            for(auto& msg : message_lists)
                            {
                                auto message_info = response->add_msg_list();
                                message_info->set_message_id(uuid());
                                message_info->set_chat_session_id(chat_session_id);
                                message_info->set_timestamp(msg.get_created_time());
                                message_info->mutable_sender()->CopyFrom(Hashmap1[msg.get_send_user_id()]);
                                switch(msg.get_message_type())
                                {
                                    case MessageContent::MessageType::STRING:
                                        message_info->mutable_message()->set_message_type(MessageType::STRING);
                                        message_info->mutable_message()->mutable_string_message()->set_content(Hashmap2[msg.get_file_id()]);
                                        break;
                                    case MessageContent::MessageType::IMAGE:
                                        message_info->mutable_message()->set_message_type(MessageType::IMAGE);
                                        message_info->mutable_message()->mutable_image_message()->set_file_id(msg.get_file_id());
                                        message_info->mutable_message()->mutable_image_message()->set_image_content(msg.content());
                                        break;
                                    case MessageContent::MessageType::FILE:
                                        message_info->mutable_message()->set_message_type(MessageType::FILE);
                                        message_info->mutable_message()->mutable_file_message()->set_file_id(msg.get_file_id());
                                        message_info->mutable_message()->mutable_file_message()->set_file_size(msg.get_file_size());
                                        message_info->mutable_message()->mutable_file_message()->set_file_name(msg.get_file_name());
                                        message_info->mutable_message()->mutable_file_message()->set_file_contents(msg.get_content());
                                        break;
                                    case MessageContent::MessageType::SPEECH:
                                        message_info->mutable_message()->set_message_type(MessageType::SPEECH);
                                        message_info->mutable_message()->mutable_speech_message()->set_file_id(msg.get_file_id());
                                        message_info->mutable_message()->mutable_speech_message()->set_file_contents(msg.get_content());
                                        break;
                                    default:
                                        Logger_ERROR("{}未能识别到的消息",rid);
                                        return errmsg_response(rid,"未能识别到的消息");
                                }
                       }
            virtual void MsgSearch(::google::protobuf::RpcController* controller,
                       const ::chatroom::MsgSearchReq* request,
                       ::chatroom::MsgSearchRsp* response,
                       ::google::protobuf::Closure* done)
                       {
                            // 1. 从请求中，获取会话 ID， 搜索关键字。
                            std::string chat_session_id = request->chat_session_id();
                            std::string search_key = request->search_key();
                            std::string rid = request->request_id();
                            // 2. 基于封装的 ES 客户端，访问 ES 服务器进行文本消息搜索（以消息内容进行搜索，
                            // 以会话 ID 进行过滤），从 ES 服务器获取到消息简息（消息 ID，会话 ID， 文本消息内
                            // 容）。
                           auto message_lists =  _es_message->search(chat_session_id,search_key);
                           if(messsge_lists.empty()){
                               response->set_success(true);
                               response->set_request_id(rid);
                               return;
                           }
                            unorder_set<std::string> set1;
                            unordered_map<std::string,UserInfo> Hashmap1;
                            for(auto& msg:message_lists)
                            {
                                set1.insert(msg.get_send_user_id());
                            }
                            bool ret = _user_info(rid,set1,Hashmap1);
                            if(!ret)
                            {
                                Logger_ERROR("{}获取用户信息失败");
                                return errmsg_response(rid,"获取用户信息失败");
                            }
                            // 4. 组织响应返回给网关服务器。
                            response->set_success(true);
                            response->set_request_id(rid);
                            for(auto& msg : message_lists)
                            {
                                auto message_info = response->add_msg_list();
                                message_info->set_message_id(uuid());
                                message_info->set_chat_session_id(chat_session_id);
                                message_info->set_timestamp(msg.get_created_time());
                                message_info->mutable_sender()->CopyFrom(Hashmap1[msg.get_send_user_id()]);
                                message_info->mutable_message()->set_message_type(MessageType::STRING);
                                message_info->mutable_message()->mutable_string_message()->set_content(Hashmap2[msg.get_file_id()]);
                            }
                       }
                       void onmessage(const char *body, size_t sz)
                       {
                            //1. 取出序列化的消息内容，进行反序列化
                            chatroom::MessageInfo Message;
                            bool ret = Message.ParseFromArray(body, sz);
                            if (ret == false) {
                                LOG_ERROR("对消费到的消息进行反序列化失败！");
                                return;
                            }
                            //2. 根据不同的消息类型进行不同的处理
                            std::string file_id, file_name, content;
                            int64_t file_size;
                            switch(Message.message().message_type())
                            {
                                case MessageContent::MessageType::STRING:
                                    //文本消息往ES中进行存储
                                    auto ret = _es_message->appendData(Message.chat_session_id(),Message.message_id(),Message.sender().user_id(),
                                    Message.timestamp(),Message.message().string_message());
                                    if(!ret)
                                    {
                                        Logger_ERROR("文本消息玩ES中存储失败!");
                                        return;
                                    }
                                    break;
                                case MessageContent::MessageType::IMAGE:
                                    bool ret = _Putfile("",Message.message().image_message().image_content(),Message.message().image_message().image_content().size(),
                                    Message.message().image_message().file_id());
                                    if(!ret)
                                    {
                                        Logger_ERROR("图片消息玩往文件子服务中存储失败!");
                                        return;
                                    }
                                    break;
                                case MessageContent::MessageType::FILE:
                                    bool ret = _Putfile(Message.message().file_message().file_id(),
                                    Message.message().file_message().file_contents(),Message.message().file_message().file_contents().size(),
                                    Message.message().file_message().file_id());
                                    if(!ret)
                                    {
                                        Logger_ERROR("文件消息玩往文件子服务中存储失败!");
                                        return;
                                    }
                                    break;
                                case MessageContent::MessageType::SPEECH:
                                    bool ret = _Putfile("",Message.message().speech_message().file_contents(),Message.message().speech_message().file_contents().size(),
                                    Message.message().speech_message().file_id());
                                    if(!ret)
                                    {
                                        Logger_ERROR("语音消息玩往文件子服务中存储失败!");
                                        return;
                                    }
                            }
                            //在将消息存储一份到数据库中
                            message msg(Message.message_id(),Message.timestamp(),Message.sender().user_id(),Message.message().message_type(),)
                            bool ret = _sql_msg->insert(msg);
                            if(!ret)
                            {
                                Logger_ERROR("消息往数据库中存储失败!");
                                return;
                            }
                       }
    private:
        //文件上传子服务 
        bool _Putfile(const std::string& filename,const std::string& file_content
        ,int64_t file_size,const std::string& file_id){
             // (1) 创建信道服务管理对象
            auto channel = _mm_channel->choose(_file_service_name);
            if(!channel)
            {
                Logger_ERROR("没有存在对应的文件子服务节点");
                return false;
            }
            chatroom::FileService_Stub stub(channel.get());
            brpc::Controller* cntl = new brpc::Controller();
            chatroom::PutSingleFileReq req;
            chatroom::PutSingleFileRsp rsp;
            auto ret = req.mutable_file_data();
            ret->set_file_name(filename);
            ret->set_file_size(file_size);
            ret->set_file_content(file_content);
            //同步调用文件上传子服务
            stub.PutSingleFile(cntl,&req,&rsp,nullptr);
            if(cntl->Failed())
            {
                Logger_ERROR("文件上传子服务调用失败,{}",cntl->ErrorText());
                delete cntl;
                return false;
            }
            file_id = rsp.file_info().file_id();
            return true;
        }
        //文件下载子服务
        bool _Getfile(const std::string& rid,unorder_set<string>& file_id,unordered_map<string,string>& Hashmap){
            // (1) 创建信道服务管理对象
            auto channel = _mm_channel->choose(_file_service_name);
            if(!channel)
            {
                Logger_ERROR("没有存在对应的文件子服务节点可以下载");
                return false;
            }
            chatroom::FileService_Stub stub(channel.get());
            brpc::Controller* cntl = new brpc::Controller();
            chatroom::GetMultiFileReq req;
            chatroom::GetMultiFileRsq rsp;
            req.set_request_id(rid);
            for(auto& id : file_id)
            {
                req.add_file_id_list(id);
            }
            //同步调用文件上传子服务
            stub.GetMultiFile(cntl,&req,&rsp,nullptr);
            if(cntl->Failed())
            {
                Logger_ERROR("文件下载子服务调用失败,{}",cntl->ErrorText());
                delete cntl;
                return false;
            }
            auto fmap = rsp.file_data();
            for(auto& tmp : fmap)
            {
                Hashmap[tmp.first] = tmp.second.file_content();
            }
            return true;
        }
        //用户子服务(获取有哪些子服务) 通过user_id来获取userinfo信息
        bool _user_info(const std::string& rid,unorder_set<string>& id_lists,unordered_map<string,UserInfo>& Hashmap){
             // (1) 创建信道服务管理对象
            auto channel = _mm_channel->choose(_file_service_name);
            if(!channel)
            {
                Logger_ERROR("没有存在对应的用户子服务节点可以查询");
                return false;
            }
            chatroom::UserService_Stub stub(channel.get());
            brpc::Controller* cntl = new brpc::Controller();
            chatroom::GetMultiUserInfoReq req;
            chatroom::GetMultiUserInfoRsq rsp;
            req.set_request_id(rid);
            for(auto& id : id_lists)
            {
                req.add_users_id(id);
            }
            //同步调用文件上传子服务
            stub.GetMultiUser(cntl,&req,&rsp,nullptr);
            if(cntl->Failed())
            {
                Logger_ERROR("用户子服务调用失败,{}",cntl->ErrorText());
                delete cntl;
                return false;
            }
            auto fmap = rsp.users_info();
            for(auto& tmp : fmap)
            {
                Hashmap[tmp.first] = tmp.second;
            }
            return true;
        }
        chatroom::MessageStorgeTable::Ptr  _sql_msg;
        chatroom::ESMessage::Ptr _es_message;
        std::string _user_service_name;
        std::string _file_service_name;
        ServiceManage::Sptr _mm_channel;
    };
    class MessageServer{
        public:
            using ptr = std::shared_ptr< MessageServer>;
            MessageServer( std::shared_ptr<elasticlient::Client> es_client,
            std::shared_ptr<odb::core::database> sql_client,
            std::string user_service_name,
            std::string file_service_name,
            std::shared_ptr<brpc::Server> server,
            chatroom::Registery::ptr rclient,
            chatroom::Discovery::ptr discover_client,
            ServiceManage::Sptr mm_channel, std::shared_ptr<MqClient> mq_client)
            :_es_client(es_client),_sql_client(sql_client),_user_service_name(user_service_name)
            ,_file_service_name(file_service_name),_server(server),_rclient(rclient),_discover_client(discover_client),_mm_channel(mm_channel)
            ,_mq_client(mq_client){}
            ~MessageServer(){}
            void start(){
                _server->RunUntilAskedToQuit();
            }
        private:
            std::shared_ptr<elasticlient::Client> _es_client;
            std::shared_ptr<odb::core::database> _sql_client;
            std::string _user_service_name;
            std::string _file_service_name;
            std::shared_ptr<brpc::Server> _server;
            chatroom::Registery::ptr _rclient;
            chatroom::Discovery::ptr _discover_client;
            ServiceManage::Sptr _mm_channel;
            std::shared_ptr<MqClient> _mq_client;
    };
    class  MessageServerBuild{
        public:
            //rpc服务
            void make_rpc_server(int32_t port,int32_t timeout,int32_t num_threads)
            {
                _server = std::make_shared<brpc::Server>();
                //3 往服务器中添加服务
                // chatroom::MessageStorgeTable::Ptr  _sql_msg;
                // chatroom::ESMessage::Ptr _es_message;
                // std::string _user_service_name;
                // std::string _file_service_name;
                // ServiceManage::Sptr _mm_channel;
                MessageServiceImp* msg = new MessageServiceImp(
                    std::make_shared<chatroom::MessageStorgeTable>(_sql_client),std::make_shared<chatroom::ESMessage>(_es_client),
                    _user_service_name,_file_service_name,_mm_channel
                );
                int ret = _server->AddService(msg,brpc::ServiceOwnership::SERVER_OWNS_SERVICE);
                if(ret == -1)
                {
                    Logger_ERROR("添加服务失败");
                    abort();
                }
                //4 启动服务器
                brpc::ServerOptions options;
                options.idle_timeout_sec = timeout; //指定超时连接时间，-1表示超时一直在连接着
                options.num_threads = num_threads;//cpu线程数量
                ret = _server->Start(port,&options);
                if(ret == -1)
                {
                    Logger_ERROR("启动服务器失败");
                    abort();
                }
                auto callback = std::bind(MessageServiceImp::onmessage,msg,std:placehoders::_1,std::placehoders::_2)
                _mq_client->Consume(_queue_name,callback);
            }
             //实例化mq客户端
            void make_MQ_client(const std::string& user,const std::string& pswd,const std::string& host,
            const std::string& exchange_name,const std::string& queue_name)
            {
                _exchange_name = exchange_name;
                _queue_name = queue_name;
                _Mq_client = std::make_shared<MqClient>(user,pswd,host);
                _Mq_client->EQbind(exchange_name,queue_name);
            }
             //实例化sql客户端
            void make_sql_client(const std::string& user,
            const std::string &pswd,
            const std::string &host,
            const std::string &db,
            const std::string &cset,
            int port,
            int conn_pool_count){
                _sql_client = chatroom::DBFactory::create(user,pswd,host,db,cset,port,conn_pool_count);
            }
            //实例化ES客户端
            void make_es_client(const std::vector<std::string> &hostUrlList){
                _es_client = chatroom::ESclientFactory::create(hostUrlList);
            }
             //实例化服务发现
            void make_discover_object(const std::string& host,std::string& basedir,std::string& fileserver_name,const std::string& user_service_name)
            {
                _mm_channel =  std::make_shared<ServiceManage>();
                _mm_channel->declare(fileserver_name);
                _mm_channel->declare(user_service_name);
                _file_service_name = fileserver_name;
                _user_service_name = user_service_name;
                auto put_cb = std::bind(&ServiceManage::ServiceOnline,_mm_channel.get(),std::placeholders::_1,std::placeholders::_2);
                auto off_cb = std::bind(&ServiceManage::ServiceOffline,_mm_channel.get(),std::placeholders::_1,std::placeholders::_2);
                _discover_client = make_shared<Discovery>(host,basedir,put_cb,off_cb);
            }
            void  make_reg_object(std::string& register_host,const std::string& hostname,std::string& access_host)
            {
                _rclient = std::make_shared<chatroom::Registery>(register_host);
                _rclient->Register(hostname,access_host);
            }
            MessageServer::ptr build()
            {
                if(!_server){
                    Logger_ERROR("RPC服务子模块没有初始化");
                    abort();
                }
                if(!_rclient){
                    Logger_ERROR("注册服务没有初始化");
                    abort();
                }
                return std::make_shared<MessageServer>( _es_client,_sql_client,_user_service_name,_file_service_name,
                    _server,_rclient,_discover_client,_mm_channel,_mq_client);
            }
        private:
            std::shared_ptr<elasticlient::Client> _es_client;
            std::shared_ptr<odb::core::database> _sql_client;
            std::string _user_service_name;
            std::string _file_service_name;
            std::shared_ptr<brpc::Server> _server;
            chatroom::Registery::ptr _rclient;
            chatroom::Discovery::ptr _discover_client;
            ServiceManage::Sptr _mm_channel;
            std::shared_ptr<MqClient> _mq_client;
            std::string _queue_name;
            std::string _exchange_name;
    };
}